package rpc_netty.Client;

import com.alibaba.nacos.api.exception.NacosException;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rpc_netty.Client.conf.ConfigureMagment_client;
import rpc_1.interfaces.ServiceRegisrty;
import rpc_1.object.RpcRequest;
import rpc_1.object.RpcResponse;
import rpc_netty.Encoder.CommonDecoder;
import rpc_netty.Encoder.CommonEncoder;
import rpc_netty.Handler.NettyClientHandler;
import rpc_netty.LoadBalancerImpl.LoadBalancerProvider;
import rpc_netty.RegistryImpl.ServiceRegistryProvider;
import rpc_netty.Serializer.KryoSerializer;
import rpc_netty.interfaces.LoadBalancer;
import rpc_netty.interfaces.RpcClient;

import java.net.InetSocketAddress;

public class NettyClient_nacos extends NettyClient implements RpcClient {
    private static final Logger logger= LoggerFactory.getLogger(NettyClient_nacos.class);

    private static final Bootstrap bootstrap;
    //直接根据sr，从nacos中获取对应的服务地址
    private static ServiceRegisrty sr;
    private InetSocketAddress cachaed_address;
    private int expireMs;
    private static long last_getAddress_time=0;
    private static LoadBalancerProvider loadBalancerProvider;
    private static final ConfigureMagment_client configureMagment=ConfigureMagment_client.getConfigureMagment_client();

    static {
        EventLoopGroup worker=new NioEventLoopGroup();
        bootstrap=new Bootstrap();
        bootstrap.group(worker)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE,true)//心跳机制
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new CommonDecoder());
                        //pipeline.addLast(new CommonEncoder(new JsonSerializer()));
                        pipeline.addLast(new CommonEncoder(1));
                        pipeline.addLast(new NettyClientHandler());
                    }
                });
    }

    //这里传入的参数，只是为了给super使用的，在这里，没什么本质用处
    //这个客户端使用的地址，是从nacos获取的，而不是这里传入的
    public NettyClient_nacos(){
        super("",0);
        //根据配置类中的信息，进行配置
        //配置注册中心类别
        String serviceRegistryType = configureMagment.getProperty("ServiceRegistryType");
        sr= ServiceRegistryProvider.getByName(serviceRegistryType);
        //读取注册中心url
        String addr = configureMagment.getProperty("ServiceRegistryUrl");
        sr.SetAdd_And_Create(addr);
        String expireTimes =  configureMagment.getProperty("ExpireTimes");
        //配置服务url缓存时间
        expireMs=Integer.parseInt(expireTimes);
        //配置负载均衡器
        String loadBalancerPolicy = configureMagment.getProperty("LoadBalancerPolicy");
        LoadBalancer loadBalancer = LoadBalancerProvider.getLoadBalancerByName(loadBalancerPolicy);
        sr.setLoadBalancer(loadBalancer);
    }

    @Override
    public Object sendRequest(RpcRequest rpcRequest) {
        try {
            //从nacos获取地址并解析
            //
            InetSocketAddress address;
            if (cachaed_address!=null&&System.currentTimeMillis()-last_getAddress_time<=expireMs){
                address=cachaed_address;
                logger.info("cached_address");
            }
            else {
                 address = sr.lookupService(rpcRequest.getInterfaceName());
                 cachaed_address=address;
                 last_getAddress_time=System.currentTimeMillis();
                 logger.info("nacos_address");
            }
            String hostName = address.getHostName();
            int port = address.getPort();
            logger.info("get service address [{}:{}] from the naocs",hostName,port);
            //根据获取到的地址，连接到服务端，得到channel
            ChannelFuture future=bootstrap.connect(hostName,port).sync();
            logger.info("connect to the server {}-{}",hostName, port);
            Channel channel = future.channel();
            if (channel!=null){
                ChannelFuture futureFinal= channel.writeAndFlush(rpcRequest);
                futureFinal.addListener(future1 -> {
                    if (future1.isSuccess()){
                        logger.info(String.format("client send the message:%s",rpcRequest.toString()));
                    }else {
                        logger.error("error when sending the message");
                        logger.error(future1.cause().getMessage());
                    }
                });
                channel.closeFuture().sync();
                AttributeKey<RpcResponse> key= AttributeKey.valueOf("rpcResponse");
                RpcResponse rpcResponse = channel.attr(key).get();
                return rpcResponse;
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            logger.error("error when sending the message",e);
        } catch (NacosException e) {
            e.printStackTrace();
        }
        return null;
    }

}
